package com.smsc.headend.task.engine.service.switchmode.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.data.entity.RdPaymentModeToPostLpData;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.task.engine.mapper.RdPaymentModeToPostLpDataMapper;
import com.smsc.headend.task.engine.service.switchmode.RdPaymentModeToPostLpDataService;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

/**
 * (业务实现)
 *
 * @author wjh
 * @date 2022/5/24 11:25.
 */
@Service
@Slf4j
public class RdPaymentModeToPostLpDataServiceImpl implements RdPaymentModeToPostLpDataService {

    @Autowired
    RdPaymentModeToPostLpDataMapper rdPaymentModeToPostLpDataMapper;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    DataSource dataSource;
    @Value("${spring.datasource.dbtype:oracle}")
    String dbType;
    @Value("${spring.datasource.dbnames}")
    String[] dnNames;

    @Override
    public void save(List<RdMeterReadsLp> rdMeterReadsLps, String taskNo) {
        List<RdPaymentModeToPostLpData> lpDataList = new ArrayList<>();
        for (RdMeterReadsLp rdMeterReadsLp : rdMeterReadsLps) {
            RdPaymentModeToPostLpData lpData = new RdPaymentModeToPostLpData();
            lpData.setLpDataItemId(rdMeterReadsLp.getLpDataItemId());
            lpData.setSubDataItemId(rdMeterReadsLp.getDataItemId());
            lpData.setMeterId(rdMeterReadsLp.getMeterId());
            lpData.setValue(rdMeterReadsLp.getValue());
            lpData.setTv(rdMeterReadsLp.getTv());
            lpData.setTaskNo(taskNo);
            lpDataList.add(lpData);
        }
        if (CollectionUtil.isNotEmpty(lpDataList)) {
            kafkaTemplate.send(TaskKafkaTopic.PAYMENT_MODE_SWITCH_TO_POST, JSONUtil.toJsonStr(lpDataList).getBytes(Charset.forName("utf-8")));
        }
    }

    @Override
    public Boolean mergeLpData(List<RdPaymentModeToPostLpData> lpDataList) {
        if (dbType.equals("oracle")) {
            return mergeForOracle(lpDataList);
        } else if (dbType.equals("postgresql")) {
            return mergeForPgsql(lpDataList);
        } else {
            log.error("merge RdPaymentModeToPostLpData unknown dbType");
        }
        return false;
    }


    private boolean mergeForOracle(List<RdPaymentModeToPostLpData> lpDataList) {
        if (CollectionUtils.isEmpty(lpDataList)) {
            return false;
        }
        log.debug("RdPaymentModeToPostLpData list size {}", lpDataList.size());

        StringBuilder sql = new StringBuilder();
        sql.append("MERGE INTO RD_PAYMENT_MODE_TO_POST_LP A")
                .append(" USING (SELECT ? AS METER_ID, ? AS TASK_NO, ? AS VALUE, ? AS TV_UPDATE, ? AS SUB_DATA_ITEM_ID, ? AS LP_DATA_ITEM_ID, ? TV FROM DUAL) B")
                .append(" ON (A.METER_ID = B.METER_ID AND A.TV = B.TV AND A.SUB_DATA_ITEM_ID = B.SUB_DATA_ITEM_ID AND A.LP_DATA_ITEM_ID=B.LP_DATA_ITEM_ID)")
                .append(" WHEN MATCHED THEN UPDATE SET A.TASK_NO = B.TASK_NO, A.VALUE = B.VALUE, A.TV_UPDATE = B.TV_UPDATE")
                .append(" WHEN NOT MATCHED THEN INSERT(METER_ID, TASK_NO, VALUE, TV_UPDATE,SUB_DATA_ITEM_ID,LP_DATA_ITEM_ID,TV)")
                .append(" VALUES(B.METER_ID, B.TASK_NO, B.VALUE, B.TV_UPDATE,B.SUB_DATA_ITEM_ID,B.LP_DATA_ITEM_ID,B.TV)");

        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);

        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql.toString())) {
            conn.setAutoCommit(false);
            for (RdPaymentModeToPostLpData lpData : lpDataList) {
                ps.setLong(1, lpData.getMeterId());
                ps.setString(2, lpData.getTaskNo());
                ps.setString(3, lpData.getValue());
                ps.setLong(4, System.currentTimeMillis() / 1000);
                ps.setLong(5, lpData.getSubDataItemId());
                ps.setLong(6, lpData.getLpDataItemId());
                ps.setDouble(7, lpData.getTv());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
            conn.close();
        } catch (SQLException e) {
            log.error("RdPaymentModeToPostLpData batch save error", e);
            return false;
        }
        return true;
    }

    private boolean mergeForPgsql(List<RdPaymentModeToPostLpData> lpDataList) {
        if (CollectionUtils.isEmpty(lpDataList)) {
            return false;
        }
        log.debug("RdPaymentModeToPostLpData list size {}", lpDataList.size());

        StringBuilder sql = new StringBuilder();


        sql.append("with upsert as ")
                .append(" (    update RD_PAYMENT_MODE_TO_POST_LP A ")
                .append("set TASK_NO = B.TASK_NO,VALUE = B.VALUE, TV_UPDATE = B.TV_UPDATE ")
                .append("from (SELECT ? AS METER_ID, ? AS TASK_NO, ? AS VALUE, ? AS TV_UPDATE, ? AS SUB_DATA_ITEM_ID, ? AS LP_DATA_ITEM_ID, ? TV) B ")
                .append(" where (A.METER_ID = B.METER_ID AND A.TV = B.TV AND A.SUB_DATA_ITEM_ID = B.SUB_DATA_ITEM_ID AND A.LP_DATA_ITEM_ID=B.LP_DATA_ITEM_ID) ")
                .append("returning A.METER_ID,A.TV,A.SUB_DATA_ITEM_ID,A.LP_DATA_ITEM_ID ) ")
                .append(" insert into RD_PAYMENT_MODE_TO_POST_LP(METER_ID, TASK_NO, VALUE, TV_UPDATE,SUB_DATA_ITEM_ID,LP_DATA_ITEM_ID,TV) ")
                .append("select * ")
                .append("from (SELECT ? AS METER_ID, ? AS TASK_NO, ? AS VALUE, ? AS TV_UPDATE, ? AS SUB_DATA_ITEM_ID, ? AS LP_DATA_ITEM_ID, ? TV) t ")
                .append("where not exists ( ")
                .append("select 1 from upsert B where t.METER_ID = B.METER_ID AND t.TV = B.TV AND t.SUB_DATA_ITEM_ID = B.SUB_DATA_ITEM_ID AND t.LP_DATA_ITEM_ID=B.LP_DATA_ITEM_ID );");
        log.debug(sql.toString());
        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);
        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql.toString())) {
            conn.setAutoCommit(false);
            for (RdPaymentModeToPostLpData lpData : lpDataList) {
                ps.setLong(1, lpData.getMeterId());
                ps.setString(2, lpData.getTaskNo());
                ps.setString(3, lpData.getValue());
                ps.setLong(4, lpData.getTvUpdate());
                ps.setLong(5, lpData.getSubDataItemId());
                ps.setLong(6, lpData.getLpDataItemId());
                ps.setDouble(7, lpData.getTv());
                ps.setLong(8, lpData.getMeterId());
                ps.setString(9, lpData.getTaskNo());
                ps.setString(10, lpData.getValue());
                ps.setLong(11, lpData.getTvUpdate());
                ps.setLong(12, lpData.getSubDataItemId());
                ps.setLong(13, lpData.getLpDataItemId());
                ps.setDouble(14, lpData.getTv());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
            conn.close();
        } catch (SQLException e) {
            log.error("RdPaymentModeToPostLpData batch save error", e);
            return false;
        }
        return true;
    }
}
